package cn.yj.frame.proxy;

import cn.yj.frame.netty.ResponseHandler;
import cn.yj.frame.protrol.RpcProtocol;
import cn.yj.register.RegisterConfig;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.timeout.IdleStateHandler;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

/**
 * Rpc调用方代理对象
 * @author 永健
 * @since 2022-01-18 17:11
 */
public class RpcReferenceProxy implements InvocationHandler {

    private Class<?> aClass;

    private String beanName;

    public RpcReferenceProxy(Class<?> aClass, String beanName) {
        this.beanName = beanName;
        this.aClass = aClass;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        ResponseHandler responseHandler = new ResponseHandler();
        EventLoopGroup group = new NioEventLoopGroup(1);
        Bootstrap bootstrap =
                new Bootstrap()
                        .group(group)
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                // 心跳监测机制
                                socketChannel.pipeline().addLast(new IdleStateHandler(3, 3, 3, TimeUnit.SECONDS));
                                // 编码处理器
                                socketChannel.pipeline().addLast("encoder", new ObjectEncoder());
                                // 解码处理器
                                socketChannel.pipeline().addLast("decoder", new ObjectDecoder(1024,
                                        ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
                                // 消息处理器
                                socketChannel.pipeline().addLast(responseHandler);
                            }
        });

        // 向注册中心拉取服务列表
        String ip = RegisterConfig.loadBalanceIp(this.beanName);
        checkIsNoExits(ip);

        // 连接提供者
        ChannelFuture connect = bootstrap.connect(ip.split(":")[0], Integer.parseInt(ip.split(":")[1])).sync();
        Channel channel = connect.channel();


        // 包装协议对象
        RpcProtocol rpcProtocol = new RpcProtocol();
        rpcProtocol.setMethodName(method.getName());
        rpcProtocol.setMethodParams(args);
        rpcProtocol.setMethodParamsType(method.getParameterTypes());
        rpcProtocol.setBeanName(this.beanName);
        // 传输
        channel.writeAndFlush(rpcProtocol);
        channel.closeFuture().sync();
        return responseHandler.getResult();
    }

    private void checkIsNoExits(String ip) {
        if (ip == null || "".equals(ip)) {
            throw new RuntimeException("服务不存在:" + this.beanName);
        }
    }
}
